package com.surfilter.massdata.es.core.service.impl;

import java.io.IOException;

import org.apache.log4j.Logger;
import org.elasticsearch.action.ShardOperationFailedException;
import org.elasticsearch.action.admin.indices.forcemerge.ForceMergeResponse;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;

import com.surfilter.massdata.es.core.client.EsClient;
import com.surfilter.massdata.es.core.client.pool.EsClientPool;
import com.surfilter.massdata.es.core.entity.EsEntity;
import com.surfilter.massdata.es.core.entity.IndexClass;
import com.surfilter.massdata.es.core.service.EsMappingService;
import com.surfilter.massdata.es.core.util.StringUtil;



/**
 * 
 * es mapping构建类
 * 
 * @author liuchen
 *
 */
public class EsMappingServiceImpl extends AbstractEsService implements EsMappingService{
	protected final Logger logger = Logger.getLogger(getClass());
	
	private EsEntity esEntity = null;
	
	private  IndexClass indexClass = null;
	
	private EsClientPool pool = null;
	
	public EsMappingServiceImpl( IndexClass indexClass,
			EsClientPool pool ,EsEntity esEntity) {
		super();
		this.esEntity = esEntity;
		this.indexClass = indexClass;
		this.pool = pool;
	}

	private void createIndex(EsClient esClient, boolean recreate,String number_of_shards,String number_of_replicas) throws IOException{
		System.out.println("createIndex number_of_shards:"+number_of_shards);
		System.out.println("createIndex number_of_replicas:"+number_of_replicas);
		boolean exists = esClient.isExists(indexClass.getIndexName());		
		if(recreate && exists) {
			esClient.deleteIndex(indexClass.getIndexName());
			logger.info("index deleted!!! indexName: " + indexClass.getIndexName());
		}
		exists = esClient.isExists(indexClass.getIndexName());
		if(!exists){
			Settings settings = Settings.builder()
					.put("index.number_of_shards", number_of_shards)
					.put("index.number_of_replicas", number_of_replicas)
					//.put("compress.default.type", "snappy")
					.put("index.compress", true)
					.put("index.store.compress.stored", true)
					.put("index.store.compress.tv", true)
					//.put("es.index.auto.create", "true")
					
					.put("index.refresh_interval", "60s")
					.put("index.translog.flush_threshold_ops", "50000")
					
					.build();
			esClient.createIndex(settings, indexClass.getIndexName());
			
			// 创建别名
			if (StringUtil.isNotNull(indexClass.getAilasName())) {
				esClient.createAlias(indexClass.getIndexName(), indexClass.getAilasName());
			}
			
			// 创建Mapping
			setMapping(esClient);
		}
	}

	private void setMapping(EsClient esClient) throws IOException {
		XContentBuilder mapping = esEntity.createMappingJson(indexClass.getTypeName());
				
		PutMappingRequest mappingRequest = Requests.putMappingRequest(indexClass.getIndexName()).type(indexClass.getTypeName())
													//.ignoreConflicts(true)
													.source(mapping);
		esClient.putMapping(mappingRequest);
	}
	
	
	/**
	 * 创建ES mapping
	 * @throws Exception 
	 * 
	 * @Description: 
	 */
	public void init(String number_of_shards,String number_of_replicas) throws Exception{
		EsClient esClient = pool.getEsClient();
		try {
			if (indexClass == null || indexClass.getIndexName() == null) {
				throw new Exception("index name is null !" );
			}
			if (indexClass.getTypeName() == null) {
				throw new Exception("type name is null !" );
			}
			createIndex(esClient, false,number_of_shards,number_of_replicas);
		} catch (Exception e) {
			logger.error("elasticsearch create mapping error !" , e);
		} finally {
			if (esClient != null) {
				pool.removeEsClient(esClient);
			}
		}
	}
	
	/**
	 * 优化索引
	 * 
	 * @Description: 
	 * @param client
	 * @param optimizeIndexName
	 */
	public void optimize(String optimizeIndexName){
		logger.info("begin optimize index:"+optimizeIndexName);
		EsClient esClient = pool.getEsClient();
		try {
			 ForceMergeResponse optResponse = esClient.optimize(optimizeIndexName);
			if (optResponse.getFailedShards() > 0) {
				ShardOperationFailedException[] excetionList = optResponse.getShardFailures();
				for (ShardOperationFailedException shardOperationFailedException : excetionList) {
					logger.error(shardOperationFailedException.toString());
				}
			}
		} catch (Exception e) {
			logger.error("elasticsearch optimize error !" , e);
		} finally {
			if (esClient != null) {
				pool.removeEsClient(esClient);
			}
		}
		logger.info("end optimize index:"+optimizeIndexName);
	}
}
